KAFKA-20253: Trigger rejoin on heartbeat thread AuthenticationException#22073
KAFKA-20253: Trigger rejoin on heartbeat thread AuthenticationException#22073ChoMinGi wants to merge 2 commits into
Conversation
The AuthenticationException catch in HeartbeatThread#run only calls setFailureCause(e) and exits, so the member stays STABLE, rejoinNeeded stays false, and heartbeatThread is cleared to null. The main thread's timeToNextHeartbeat() then returns 0 from the stale heartbeatTimer, causing NetworkClient.poll(0) to spin via selectNow() indefinitely. Extend the existing "error -> requestRejoin()" pattern already used in HeartbeatResponseHandler to the auth catch, so the next poll() runs ensureActiveGroup() -> startHeartbeatThreadIfNeeded() -> heartbeat.resetTimeouts(), restoring consumer state. Verified with a new unit test and end-to-end via mstruk/kafka-consumer-reproducer (101% -> ~0.3% CPU, fresh generation).
github-actions
Bot
added
triage
|
A label of 'needs-attention' was automatically added to this PR in order to raise the |
|
A label of 'needs-attention' was automatically added to this PR in order to raise the |
|
I tested this PR with the reproducer, and I was unable to trigger the issue. It looks like this fix works fine. |
|
@kirktrue @FrankYang0529 Can you take a look? Thanks |
|
A label of 'needs-attention' was automatically added to this PR in order to raise the |
ezhou413
left a comment
ezhou413
left a comment
There was a problem hiding this comment.
LGTM.
What I did:
- I ran the reproducer with these changes and wasn't able to get a busy CPU after a few tries
- Dug into the code and found that after we get an
AuthenticationException, we need to recreate the heartbeat thread, for which there is only 1 code path, triggered by a rejoin and called byAbstractCoordinator.ensureActiveGroup. Since a rejoin does a lot of things, I looked into creating another code path that only recreates the heartbeat thread, but found that it's non trivial, and would only help the case where we hit an auth outage that is shorter than the consumer's session timeout, which is rare. All of this means that if we hit anAuthenticationException, we most likely need a rejoin anyways
AndrewJSchofield
left a comment
AndrewJSchofield
left a comment
There was a problem hiding this comment.
Thanks for the PR. I think it would be worth validating in the test that clearing the authentication problem does actually allow the consumer to join the group without any other side-effects. I don't think requestRejoin was really quite intended for this use, but I do also think that it does the job. That's why I'd like the test to be a bit more comprehensive.
| } | ||
|
|
||
| @Test | ||
| public void testAuthenticationErrorInHeartbeatThreadTriggersRejoin() throws Exception { |
There was a problem hiding this comment.
This test in a way tests that the chosen fix was implemented, namely that rejoin was requested. What I'd really like to see is a test that does an authentication failure on the first heartbeat, then the rejoin is requested, and a subsequent successful heartbeat works as expected.
There was a problem hiding this comment.
Done in bdc9755 — The test now covers the full cycle.
auth failure on the first heartbeat, rejoin requested, successful JoinGroup/SyncGroup once auth recovers, then a subsequent heartbeat completing without another rejoin.
To keep the last assertion from passing vacuously, the test requires the prepared heartbeat response to be consumed with no heartbeat left in flight — the matcher only accepts a HeartbeatRequest, so the post-recovery heartbeat must actually have been sent and processed.
…ry cycle - Verify rejoin completes after authentication recovers - Verify a post-recovery heartbeat succeeds without another rejoin - Ensure the prepared heartbeat response is consumed
|
Thanks for the reviews :) On requestRejoin not being quite intended for this — agreed. I did consider restarting only the heartbeat thread. But the client can't know upfront whether the membership is still valid, so that path would need its own fallback to a rejoin when the session has expired, effectively duplicating recovery logic. Reusing the proven rejoin path felt like the right scope for this fix, even though it can cost an unnecessary rebalance when the outage is shorter than the session timeout. If there's interest, I'd be happy to explore a dedicated heartbeat-recovery path in a follow-up. |
5 participants
KAFKA-20253
Reproduced the 100% CPU busy loop using @mstruk's
reproducer and
tracked it down to
AbstractCoordinator.HeartbeatThread. When anAuthenticationExceptionhits the outer catch (e.g. after a transientOAuth server outage), it only calls
setFailureCause(e)and exits — butnothing triggers a rejoin. So
statestaysSTABLE,rejoinNeededstays
false,heartbeatThreadgets cleared tonull, andtimeToNextHeartbeat()returns the staleheartbeatTimer.remainingMs() == 0. The main thread then spinsNetworkClient.poll(0)viaselectNow()indefinitely.Fixed by adding
requestRejoin()to the auth catch block — the samepattern already used for
REBALANCE_IN_PROGRESS/UNKNOWN_MEMBER_IDin
HeartbeatResponseHandler. On the nextpoll()this kicks offensureActiveGroup()→startHeartbeatThreadIfNeeded()→heartbeat.resetTimeouts(), and the consumer recovers on its own.GroupAuthorizationExceptionis left as-is since rejoining on an ACLdenial would just loop.
Unit test and validation with the reproducer:
Thanks to @mstruk for the detailed report and the reproducer.
This change targets the Classic consumer; the related AsyncKafkaConsumer issue is handled separately in #21714.
Reviewers: Evan Zhou ezhou@confluent.io, Andrew Schofield aschofield@confluent.io